package com.lvy.hczn.mq.rocket;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
@Slf4j
public class ProducerUtil {

    @Autowired
    private DefaultMQProducer defaultProducer;

    @Autowired
    private TransactionMQProducer transactionMQProducer;

    /**
     * 同步发送消息
     *
     * @param msgTag      标签，可用于消息小分类标注
     * @param messageBody 消息body内容，生产者自定义内容
     * @param msgKey      消息key值，建议设置全局唯一，可不传，不影响消息投递
     * @return success:SendResult or error:null
     */
    public SendResult sendMsg(String topic, String msgTag, byte[] messageBody, String msgKey) throws Exception {
        Message msg = new Message(topic, msgTag, msgKey, messageBody);
        return this.send(msg, Boolean.FALSE);
    }

    /**
     * 同步发送定时/延时消息
     *
     * @param msgTag      标签，可用于消息小分类标注，对消息进行再归类
     * @param messageBody 消息body内容，生产者自定义内容，二进制形式的数据
     * @param msgKey      消息key值，建议设置全局唯一值，可不设置，不影响消息收发
     * @param delayTime   服务端发送消息时间，立即发送输入0或比更早的时间
     * @return success:SendResult or error:null
     */
    public SendResult sendTimeMsg(String topic, String msgTag, byte[] messageBody, String msgKey, int delayTime) throws Exception {
        Message msg = new Message(topic, msgTag, msgKey, messageBody);
        msg.setDelayTimeLevel(delayTime);
        return this.send(msg, Boolean.FALSE);
    }

    /**
     * 发送单向消息
     *
     * @param topic
     * @param msgTag
     * @param messageBody
     * @param msgKey
     * @return
     * @author yw
     * @date 2021-06-11 14:03:51
     **/
    public void sendOneWayMsg(String topic, String msgTag, byte[] messageBody, String msgKey) throws Exception {
        Message msg = new Message(topic, msgTag, msgKey, messageBody);
        this.send(msg, Boolean.TRUE);
    }

    /**
     * 普通消息发送发放
     *
     * @param msg
     * @param isOneWay 是否单向发送
     * @return
     * @author yw
     * @date 2021-06-11 14:03:25
     **/
    private SendResult send(Message msg, Boolean isOneWay) throws Exception {
        if (isOneWay) {
            //由于在 oneway 方式发送消息时没有请求应答处理，一旦出现消息发送失败，则会因为没有重试而导致数据丢失。
            //若数据不可丢，建议选用同步或异步发送方式。
            defaultProducer.sendOneway(msg);
            success(msg, "单向消息MsgId不返回");
            return null;
        } else {
            //可靠同步发送
            SendResult sendResult = defaultProducer.send(msg);
            //获取发送结果，不抛异常即发送成功
            if (sendResult != null) {
                success(msg, sendResult.getMsgId());
                return sendResult;
            } else {
                error(msg, null);
                return null;
            }
        }
    }

    //对于使用异步接口，可设置单独的回调处理线程池，拥有更灵活的配置和监控能力。
    //根据项目需要，服务器配置合理设置线程数，线程太多有OOM 风险，
    private ExecutorService threads = Executors.newFixedThreadPool(3);
    //仅建议执行轻量级的Callback任务，避免阻塞公共线程池 引起其它链路超时。

    /**
     * 异步发送普通消息
     *
     * @param topic
     * @param msgTag
     * @param messageBody
     * @param msgKey
     * @return
     * @author yw
     * @date 2021-06-11 14:02:38
     **/
    public void sendAsyncMsg(String topic, String msgTag, byte[] messageBody, String msgKey) throws Exception {
        defaultProducer.setCallbackExecutor(threads);
        Message msg = new Message(topic, msgTag, msgKey, messageBody);
        defaultProducer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                if (sendResult != null) {
                    success(msg, sendResult.getMsgId());
                } else {
                    log.error("异步消息sendAsyncMsg返回NULL");
                }
            }

            @Override
            public void onException(Throwable e) {
                //出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。
                error(msg, e.getMessage());
            }
        });
    }

    /**
     * 发送批量消息
     * 批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic，
     * 相同的waitStoreMsgOK，而且不能是延时消息。此外，这一批消息的总大小不应超过4MB。
     *
     * @param list
     * @return
     * @author yw
     * @date 2021-06-11 14:03:25
     **/
    public SendResult sendBatch(List<Message> list) throws Exception {
        SendResult sendResult = defaultProducer.send(list);
        //获取发送结果，不抛异常即发送成功
        if (sendResult != null) {
            for (Message msg : list) {
                success(msg, sendResult.getMsgId());
            }
            return sendResult;
        } else {
            for (Message msg : list) {
                error(msg, null);
            }
            return null;
        }
    }


    /**
     * 失败
     *
     * @param msg
     * @param exceptionMsg
     * @return void
     * @author yw
     * @date 2021-06-11 19:36:10
     */
    private void error(Message msg, String exceptionMsg) {
        log.error("发送MQ消息失败-- Message:{}", msg.toString());
        log.error("errorMsg --- {}", exceptionMsg);
    }

    /**
     * 成功
     *
     * @param msg
     * @param messageId
     * @return
     * @author yw
     * @date 2021-06-11 14:02:02
     **/
    private void success(Message msg, String messageId) {
        log.info("发送消息成功 -- messageId:{},Message:{}", messageId, msg.toString());
    }

    /**
     * mq 事物生产者
     *
     * @param topic
     * @param msgTag
     * @param messageBody
     * @param msgKey
     * @return
     * @author yw
     * @date 2021-06-11 14:01:31
     **/
    public TransactionSendResult sendTransactionMQProducer(String topic, String msgTag, byte[] messageBody, String msgKey) throws Exception {
        Message msg = new Message(topic, msgTag, msgKey, messageBody);
        return transactionMQProducer.sendMessageInTransaction(msg, null);
    }

}
